package com.xiguanlezz.cn.client.naming.remote.grpc;

import com.xiguanlezz.cn.api.naming.pojo.Instance;
import com.xiguanlezz.cn.api.naming.pojo.ServiceInfo;
import com.xiguanlezz.cn.api.remote.NamingRemoteConstants;
import com.xiguanlezz.cn.api.remote.RemoteConstants;
import com.xiguanlezz.cn.api.remote.request.AbstractNamingRequest;
import com.xiguanlezz.cn.api.remote.request.InstanceRequest;
import com.xiguanlezz.cn.api.remote.request.SubscribeServiceRequest;
import com.xiguanlezz.cn.api.remote.response.Response;
import com.xiguanlezz.cn.api.remote.response.ResponseCode;
import com.xiguanlezz.cn.api.remote.response.SubscribeServiceResponse;
import com.xiguanlezz.cn.client.naming.core.ServerListManager;
import com.xiguanlezz.cn.client.env.NacosClientProperties;
import com.xiguanlezz.cn.client.naming.cache.ServiceInfoHolder;
import com.xiguanlezz.cn.client.naming.remote.NamingClientProxy;
import com.xiguanlezz.cn.client.naming.remote.grpc.redo.NamingGrpcRedoService;
import com.xiguanlezz.cn.client.naming.remote.grpc.redo.data.InstanceRedoData;
import com.xiguanlezz.cn.common.remote.ConnectionType;
import com.xiguanlezz.cn.common.remote.RpcClientFactory;
import com.xiguanlezz.cn.common.remote.client.RpcClient;

import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;

/**
 * @Author：习惯了种种
 * @CreateDate：2024/11/16 17:39
 * @Description：GRPC通信对象实现类
 */
public class NamingGrpcClientProxy implements NamingClientProxy {
    private String namespace; /* 命名空间 */
    private RpcClient rpcClient;
    private String uuid;

    // 断线重连后的重试操作服务对象
    private final NamingGrpcRedoService redoService;


    public NamingGrpcClientProxy(String namespace, ServerListManager serverListManager, NacosClientProperties nacosClientProperties) {
        this.uuid = UUID.randomUUID().toString();
        this.namespace = namespace;
        this.rpcClient = RpcClientFactory.createClient(uuid, ConnectionType.GRPC, getLabels());
        // 初始化重试操作服务对象
        this.redoService = new NamingGrpcRedoService(this, nacosClientProperties);
        start(serverListManager, null);
    }

    public NamingGrpcClientProxy(String namespace, ServerListManager serverListManager, ServiceInfoHolder serviceInfoHolder, NacosClientProperties nacosClientProperties) {
        this.uuid = UUID.randomUUID().toString();
        this.namespace = namespace;
        this.rpcClient = RpcClientFactory.createClient(uuid, ConnectionType.GRPC, getLabels());
        // 初始化重试操作服务对象
        this.redoService = new NamingGrpcRedoService(this, nacosClientProperties);
        start(serverListManager, serviceInfoHolder);
    }

    private void start(ServerListManager serverListManager, ServiceInfoHolder serviceInfoHolder) {
        // 设置Nacos服务端地址管理器，同时将RPC客户端状态从等待初始化改为初始化状态
        rpcClient.serverListFactory(serverListManager);
        // 将重试服务对象注册到事件监听器中
        rpcClient.registerConnectionListener(redoService);
        if (Objects.nonNull(serviceInfoHolder)) {
            // 注册服务端主动推送服务实例信息的请求处理器
            rpcClient.registerServerRequestHandler(new NamingPushRequestHandler(serviceInfoHolder));
        }
        // 启动RPC客户端，说白了就是和服务端建立连接
        rpcClient.start();
    }

    /**
     * 用于扩展的labels
     */
    private Map<String, String> getLabels() {
        Map<String, String> labels = new HashMap<>();
        labels.put(RemoteConstants.LABEL_SOURCE, RemoteConstants.LABEL_SOURCE_SDK);
        labels.put(RemoteConstants.LABEL_MODULE, RemoteConstants.LABEL_MODULE_NAMING);
        return labels;
    }

    @Override
    public void registerService(String serviceName, String groupName, Instance instance) throws Exception {
        if (instance.isEphemeral()) {
            // 临时实例的注册
            registerServiceForEphemeral(serviceName, groupName, instance);
        } else {
            // 持久化实例的注册
            // TODO(xiguanlezz)：后面实现持久化实例的注册逻辑
        }
    }

    private void registerServiceForEphemeral(String serviceName, String groupName, Instance instance) throws Exception {
        doRegisterService(serviceName, groupName, instance);
    }

    private void doRegisterService(String serviceName, String groupName, Instance instance) throws Exception {
        // 未显示传递expectedRegistered时，默认值为true，表示此时是服务注册状态
        InstanceRedoData redoData = InstanceRedoData.build(serviceName, groupName, instance);
        // 缓存一个重试对象，最终期望状态是注册，此时registered为false，表示还未注册
        redoService.cacheInstanceForRedo(serviceName, groupName, instance);
        InstanceRequest instanceRequest = new InstanceRequest(
                namespace, serviceName, groupName,
                NamingRemoteConstants.REGISTER_INSTANCE, instance
        );
        // 向服务端发送实例注册请求，并不关注响应结果，因为其中带有重试逻辑，多次不成功会有定时任务扫描重试对象map并进行重试
        // 有点最大努力尝试成功的感觉。。。
        requestToServer(instanceRequest, Response.class);
        // 将registered设置为true，表示已经注册
        redoService.instanceRegistered(serviceName, groupName);
    }

    @Override
    public void deregisterService(String serviceName, String groupName, Instance instance) throws Exception {
        if (instance.isEphemeral()) {
            deregisterServiceForEphemeral(serviceName, groupName, instance);
        } else {
            // 持久化实例的注销
            // TODO(xiguanlezz)：后面实现持久化实例的注销逻辑
        }
    }

    private void deregisterServiceForEphemeral(String serviceName, String groupName, Instance instance) throws Exception {
        // 服务注销的前提是服务已经注册到Nacos服务端了
        // 此时将缓存map中的重试对象的期望状态改为注销，并设置unregistering为true
        redoService.instanceDeregister(serviceName, groupName);
        doDeregisterService(serviceName, groupName, instance);
    }

    private void doDeregisterService(String serviceName, String groupName, Instance instance) throws Exception {
        InstanceRequest request = new InstanceRequest(
                namespace, serviceName, groupName,
                NamingRemoteConstants.DE_REGISTER_INSTANCE, instance
        );
        // 向服务端发送实例注销请求，并不关注响应结果，因为其中带有重试逻辑，多次不成功会有定时任务扫描重试对象map并进行重试
        requestToServer(request, Response.class);
        // 将registered设置为false，表示已经注销
        redoService.instanceDeregistered(serviceName, groupName);
    }

    @Override
    public ServiceInfo subscribe(String serviceName, String groupName, String clusters) throws Exception {
        // 创建一个订阅操作的重试对象，类似服务注册
        redoService.cacheSubscriberForRedo(serviceName, groupName, clusters);
        return doSubscribe(serviceName, groupName, clusters);
    }

    public ServiceInfo doSubscribe(String serviceName, String groupName, String clusters) throws Exception {
        SubscribeServiceRequest request = new SubscribeServiceRequest(
                namespace, groupName, serviceName, clusters,
                true
        );
        // 向服务端发送订阅实例请求，服务端会把订阅到的服务实例返回，因为其中带有重试逻辑，多次不成功会有定时任务扫描重试对象map并进行重试
        SubscribeServiceResponse response = requestToServer(request, SubscribeServiceResponse.class);
        // 将registered设置为true，表示已经订阅
        redoService.subscriberRegistered(serviceName, groupName, clusters);
        // 返回得到的服务实例信息
        return response.getServiceInfo();
    }

    @Override
    public void unsubscribe(String serviceName, String groupName, String clusters) throws Exception {
        // 取消订阅实例的前提是客户端已经订阅到实例了
        // 此时将缓存map中的重试对象的期望状态改为取消订阅，并设置unregistering为true
        redoService.subscriberDeregister(serviceName, groupName, clusters);
        doUnsubscribe(serviceName, groupName, clusters);
    }

    @Override
    public boolean isSubscribed(String serviceName, String groupName, String clusters) throws Exception {
        return redoService.isSubscriberRegistered(serviceName, groupName, clusters);
    }

    public void doUnsubscribe(String serviceName, String groupName, String clusters) throws Exception {
        SubscribeServiceRequest request = new SubscribeServiceRequest(
                namespace, groupName, serviceName, clusters,
                false
        );
        // 向服务端发送取消订阅实例请求，并不关注响应结果，因为其中带有重试逻辑，多次不成功会有定时任务扫描重试对象map并进行重试
        requestToServer(request, SubscribeServiceResponse.class);
        // 移除本地缓存中的订阅数据
        redoService.removeSubscriberForRedo(serviceName, groupName, clusters);
    }

    private <T extends Response> T requestToServer(AbstractNamingRequest request, Class<T> responseClazz) throws Exception {
        try {
            Response response = rpcClient.request(request);
            if (response.getResultCode() != ResponseCode.SUCCESS.getCode()) {
                throw new RuntimeException("Response is not successful, msg: " + response.getMessage());
            }
            if (Response.class.isAssignableFrom(responseClazz)) {
                return (T) response;
            }
            throw new RuntimeException("Server return invalid response");
        } catch (Exception e) {
            throw e;
        }
    }

    public boolean isEnable() {
        return rpcClient.isRunning();
    }
}
